package com.yao.controller;

import com.yao.entity.User;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author YaoXJ
 * @Date 2023-07-21
 * @Description
 */
@RestController
@Slf4j
public class ProducerController {

    @Value(value = "${rocketmq.producer.topic}")
    private String topic;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
    private String syncTag;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.async-tag}")
    private String asyncag;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.oneway-tag}")
    private String onewayTag;

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    /**
     * 普通消息
     * 发送方式一：send
     * @return
     */
    @GetMapping("/send")
    public String send() throws Exception {
        // 构建消息
        String message = "普通消息send";
        Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
        //必须使用RocketMQ包中Message类型
        SendResult send = rocketMQTemplate.getProducer().send(msg);
        System.out.println(send);
        return send.toString();
    }

    /**
     * 普通消息
     * 发送方式二：convertAndSend
     */
    @RequestMapping("/convertAndSend")
    public String sendMessage() {
        // convertAndSend可以传任意对象
        rocketMQTemplate.convertAndSend(topic, "消息类型：convertAndSend");
        return "消息已发送";
    }

    /**
     * 过滤消息
     */
    @GetMapping("/tagSend")
    public String tagSend() throws Exception {
        // 构建消息
        String message = "过滤消息";
        Message msg = new Message(topic, "abc",message.getBytes(StandardCharsets.UTF_8));
        //必须使用RocketMQ包中Message类型
        SendResult send = rocketMQTemplate.getProducer().send(msg);
        System.out.println(send);
        return send.toString();
    }

    /**
     * 同步消息
     */
    @RequestMapping("/syncSend")
    public String pushMessage() {
        // 构建消息
        String message = "同步消息";
        Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
        // 设置发送地和消息信息并发送同步消息
        SendResult sendResult = rocketMQTemplate.syncSend(syncTag, message);
        // 解析发送结果
        if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
            System.out.println("发送成功");
        }
        return sendResult.toString();
    }

    /**
     * 异步消息
     */
    @RequestMapping("/asyncSend")
    public String pushAsyncMessage() {
        // 构建消息
        String message = "异步消息";
        Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
        // 设置发送地和消息信息并发送异步消息
        rocketMQTemplate.asyncSend(asyncag, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("异步消息发送失败");
            }
        });
        return "异步消息已发送";
    }

    /**
     * 单向消息
     */
    @GetMapping("sendOneWay")
    public String sendOneWay() {
        User user = new User("cc", 19);
        rocketMQTemplate.sendOneWay(onewayTag, user);
        return "单向消息发送成功";
    }

    /**
     * 延时消息
     */
    @GetMapping("sendDelayLevel")
    public String sendDelayLevel() {
        User user = new User("cc", 19);
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(user).build(), 2000, 3);
        return "延时消息发送成功";
    }

    /**
     * 批量消息
     */
    @GetMapping("sendMatch")
    public String sendMatch() {
        List<User> users= new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            User user = new User("小" + i, i);
            users.add(user);
        }
        rocketMQTemplate.syncSend(topic, users, 1000);
        return "批量发送成功";
    }
}
